package com.atguigu.flinkSql;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.descriptors.Json;
import org.apache.flink.table.descriptors.Kafka;
import org.apache.flink.table.descriptors.Schema;

import static org.apache.flink.table.api.Expressions.$;

/**
 * @author wky
 * @create 2021-07-21-11:40
 */
public class Flink04_TableApi_Kafka_source {
    public static void main(String[] args) {
        StreamExecutionEnvironment environment = StreamExecutionEnvironment.getExecutionEnvironment();
        environment.setParallelism(1);
        StreamTableEnvironment tableEnvironment = StreamTableEnvironment.create(environment);
//元数据
        Schema schema = new Schema()
                .field("id", DataTypes.STRING())
                .field("ts", DataTypes.BIGINT())
                .field("vc", DataTypes.INT());
//连接文件 创建临时表
        tableEnvironment.connect(new Kafka()
                //通用版本
                .version("universal")
                .topic("sensor")
                //读取方式 最新读取
                .startFromLatest()
                .property("group.id","bigdata")
                //服务
                .property("bootstrap.servers","hadoop102:9092,hadoop103:9092,hadoop104:9092")
        ).withFormat(new Json())
        .withSchema(schema)
        .createTemporaryTable("sensor");

        //读取数据
        TableResult result = tableEnvironment.from("sensor")
                .groupBy($("id"))
                .select($("id"), $("id").count())
                .execute();
        result.print();

    }
}
